#include "ipc_server.h"

#include <arpa/inet.h>
#include <cstdio>
#include <mutex>
#include <poll.h>
#include <sys/eventfd.h>
#include <unistd.h>
#include <vector>
#include <cstring>

#include <logger.h>
#include <ipc_message_parcel.h>

#include "ipc_service_registry.h"

const sptr<IpcServer> &IpcServer::GetInstance()
{
    if (instance == nullptr) {
        static std::mutex mutex;
        std::lock_guard<std::mutex> lock(mutex);
        if (instance == nullptr) {
            instance = new IpcServer();
        }
    }
    return instance;
}

int IpcServer::Start(int port)
{
    socketFd = socket(AF_INET, SOCK_STREAM, 0);
    if (socketFd < 0) {
        LOG2SE(ERROR) << "socket" << LOGSYSERR;
        return -1;
    }

    struct sockaddr_in addr = {
        .sin_family = AF_INET,
        .sin_port = htons(port),
        .sin_addr = {
            .s_addr = htonl(INADDR_ANY),
        },
    };
    auto ret = bind(socketFd, (struct sockaddr *)&addr, sizeof(struct sockaddr));
    if (ret < 0) {
        LOG2SE(ERROR) << "bind" << LOGSYSERR;
        return ret;
    }
    LOG2SO(INFO) << "bind success";

    constexpr int requestQueueSize = 20;
    ret = listen(socketFd, requestQueueSize);
    if (ret < 0) {
        LOG2SO(INFO) << "listen" << LOGSYSERR;
        return ret;
    }

    LOG2SO(INFO) << "listen success";
    return ret;
}

void IpcServer::SetRegistry(IpcServiceRegistry &registry)
{
    registry_ = &registry;
}

void IpcServer::Run()
{
    interruptFd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
    pollfds = {
        { .fd = socketFd, .events = POLLIN, },
        { .fd = interruptFd, .events = POLLIN, },
    };
    while (true) {
        std::stringstream ss;
        ss << "poll " << pollfds.size() << " fds";
        LOGSCOPED(scoped, LOG2SO(DEBUG), ss.str());
        for (const auto &pollfd : pollfds) {
            LOG2SO(DEBUG) << pollfd.fd;
        }

        int32_t ret = 0;
        do {
            ret = poll(pollfds.data(), pollfds.size(), -1);
        } while (ret == -1 && errno == EINTR);

        if (ret < 0) {
            LOG2SE(ERROR) << "poll" << LOGSYSERR;
            break;
        }

        if (ProcessPollFds()) {
            break;
        }
    }

    for (const auto &pollfd : pollfds) {
        close(pollfd.fd);
    }
}

void IpcServer::Stop()
{
    if (interruptFd < 0) {
        return;
    }

    LOG2SO(INFO) << "stopping";
    uint64_t buf = 1;
    int32_t ret = 0;
    do {
        ret = write(interruptFd, &buf, sizeof(buf));
    } while (ret == -1 && errno == EINTR);
}

int IpcServer::Send(IpcMessageParcel &parcel, struct IpcClient client)
{
    auto it = clientfds.find(client);
    if (it == clientfds.end()) {
        LOG2SE(ERROR) << "cannot found client{" << client.pid << ":" << client.cid << "}";
        return -1;
    }
    LOG2SO(DEBUG) << "send way: " << parcel.way << ", pid: " << client.pid << ", cid: " << client.cid << ", data: " << parcel.data;
    return parcel.Write(it->second);
}

std::ostream &operator <<(std::ostream &os, const struct sockaddr_in &addr)
{
    os << ((addr.sin_addr.s_addr >>  0) & 0xff) << "."
       << ((addr.sin_addr.s_addr >>  8) & 0xff) << "."
       << ((addr.sin_addr.s_addr >> 16) & 0xff) << "."
       << ((addr.sin_addr.s_addr >> 24) & 0xff) << ":"
       << addr.sin_port;
    return os;
}

int IpcServer::ProcessPollFds()
{
    LOGSCOPED(scoped, LOG2SO(DEBUG));
    std::vector<int> addConnectionFds;
    for (auto it = pollfds.begin(); it != pollfds.end(); it++) {
        auto &pollfd = *it;
        if (!(pollfd.revents & POLLIN)) {
            continue;
        }

        std::stringstream ss;
        ss << "next fd: " << pollfd.fd;
        LOGSCOPED(scoped, LOG2SO(DEBUG), ss.str());
        if (pollfd.fd == socketFd) {
            struct sockaddr_in clientAddr;
            socklen_t addrlen = sizeof(sockaddr);
            LOG2SO(DEBUG) << "accepting";
            auto connFd = accept(socketFd, (struct sockaddr *)&clientAddr, &addrlen);
            addConnectionFds.push_back(connFd);
            LOG2SO(DEBUG) << "accept " << connFd << ", from " << clientAddr;
        } else if (pollfd.fd == interruptFd) {
            LOG2SO(DEBUG) << "interrupt";
            return 1;
        } else {
            auto &connFd = pollfd.fd;

            IpcMessageParcel parcel;
            auto size = parcel.Read(connFd);

            if (size > 0) {
                clientfds[parcel] = connFd;
                registry_->DispatchMessage(parcel);
            } else if (size == 0) {
                pollfds.erase(it--);
                clientfds.erase(parcel);
            } else {
                LOG2SE(ERROR) << LOGSYSERR;
            }
        }
    }

    for (auto connFd : addConnectionFds) {
        pollfds.push_back({.fd = connFd, .events = POLLIN});
    }
    return 0;
}
